package com.gy.flink.state;

import lombok.Data;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

public class StreamJoinListStateMain {


    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        DataStreamSource<String> ds1 = env.addSource(new OrderSource("./flink.input/order1.txt"));
        DataStreamSource<String> ds2 = env.addSource(new OrderSource("./flink.input/order2.txt"));

        SingleOutputStreamOperator<OrderInfo1> orderInfoStream1 = ds1.map((MapFunction<String, OrderInfo1>) OrderInfo1::build);
        SingleOutputStreamOperator<OrderInfo2> orderInfoStream2 = ds2.map((MapFunction<String, OrderInfo2>) OrderInfo2::build);

        KeyedStream<OrderInfo1, String> keyByOrderStream1 = orderInfoStream1.keyBy(OrderInfo1::getNum);
        KeyedStream<OrderInfo2, String> keyByOrderStream2 = orderInfoStream2.keyBy(OrderInfo2::getNum);

        ConnectedStreams<OrderInfo1, OrderInfo2> connectStream = keyByOrderStream1.connect(keyByOrderStream2);


        connectStream.flatMap(new OrderJoinFlatMapListState()).print().setParallelism(1);


        env.execute(StreamJoinListStateMain.class.getSimpleName());
    }


    @Data
    static class OrderInfo1 {
        private String num;
        //订单号,购买的商品,商品的价格
        private String name;
        private Double price;

        static OrderInfo1 build(String line) {
            if (line == null) return null;
            String[] split = line.split(",");
            OrderInfo1 bean = new OrderInfo1();
            bean.setNum(split[0]);
            bean.setName(split[1]);
            bean.setPrice(Double.valueOf(split[2]));
            return bean;
        }
    }

    @Data
    public static class OrderInfo2 {
        //订单号,下单的地点,下单时间
        private String num;
        private String address;
        private String cTime;


        static OrderInfo2 build(String line) {
            if (line == null) return null;
            String[] split = line.split(",");
            OrderInfo2 bean = new OrderInfo2();
            bean.setNum(split[0]);
            bean.setAddress(split[1]);
            bean.setCTime(split[2]);
            return bean;
        }
    }

    public static class OrderSource implements SourceFunction<String> {

        private String fileName;
        private BufferedReader br;

        public OrderSource(String fileName) {
            this.fileName = fileName;
        }

        @Override
        public void run(SourceContext<String> sc) throws Exception {
            br = new BufferedReader(new InputStreamReader(new FileInputStream(fileName)));
            String line = null;
            while ((line = br.readLine()) != null) {
                sc.collect(line);
                TimeUnit.MICROSECONDS.sleep(5000);
            }
        }

        @Override
        public void cancel() {
            try {
                if (br != null)
                    br.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static class OrderJoinFlatMapListState extends RichCoFlatMapFunction<OrderInfo1, OrderInfo2, Tuple2<OrderInfo1, OrderInfo2>> {

        ListState<OrderInfo1> state1;
        ListState<OrderInfo2> state2;

        @Override
        public void open(Configuration parameters) throws Exception {
            ListStateDescriptor<OrderInfo1> descriptor1 = new ListStateDescriptor<>(
                    "order1",
                    Types.GENERIC(OrderInfo1.class)
            );
            ListStateDescriptor<OrderInfo2> descriptor2 = new ListStateDescriptor<>(
                    "order2",
                    Types.GENERIC(OrderInfo2.class)
            );
            state1 = super.getRuntimeContext().getListState(descriptor1);
            state2 = super.getRuntimeContext().getListState(descriptor2);
        }

        @Override
        public void flatMap1(OrderInfo1 orderInfo1, Collector<Tuple2<OrderInfo1, OrderInfo2>> out) throws Exception {
            Iterable<OrderInfo2> state2Value = state2.get();
            if (!Lists.newArrayList(state2Value).isEmpty()) {
                Iterator<OrderInfo2> it = state2Value.iterator();
                while (it.hasNext()) {
                    out.collect(Tuple2.of(orderInfo1, it.next()));
                    it.remove();
                }
            } else {
                state1.add(orderInfo1);
            }

        }

        @Override
        public void flatMap2(OrderInfo2 orderInfo2, Collector<Tuple2<OrderInfo1, OrderInfo2>> out) throws Exception {
            Iterable<OrderInfo1> state1Value = state1.get();
            ArrayList<OrderInfo1> list = Lists.newArrayList(state1Value);
            if (!list.isEmpty()) {
                Iterator<OrderInfo1> it = state1Value.iterator();
                while (it.hasNext()) {
                    out.collect(Tuple2.of(it.next(), orderInfo2));
                    it.remove();
                }
            } else {
                state2.add(orderInfo2);
            }
        }
    }
}
